package io.sqrtqiezi.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object WindowDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("spark socket streaming")
      .setMaster("local[*]")

    val ssc = new StreamingContext(conf, Seconds(5))
    ssc.sparkContext.setLogLevel("error")

    val lines: ReceiverInputDStream[String] =
      ssc.socketTextStream("localhost", 1521)

    lines.foreachRDD{ (rdd, time) =>
      println(s"rdd = ${rdd.id}; time = $time")
      rdd.foreach(println)
    }

    val res1 = lines.reduceByWindow(_ + " " + _, Seconds(10), Seconds(5))
    res1.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
